package com.atguigu.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TableProcess;
import com.atguigu.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;

public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private Connection connection;

    public TableProcessFunction(MapStateDescriptor mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    @Override
    public void processElement(JSONObject value, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        // 获取状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        TableProcess tableProcess = broadcastState.get(value.getString("table"));

        if (tableProcess != null){
            // 过滤字段
            filterColumns(value.getJSONObject("data"), tableProcess.getSinkColumns());

            // 把目标表，写入数据
            if (!"delete".equals(value.getString("type"))) {
                value.put("sinkTable", tableProcess.getSinkTable());
                out.collect(value);
            }
        }else {
            System.out.println("Key: " + value.getString("table") + " 不存在！");
        }
    }

    private void filterColumns(JSONObject data, String sinkColumns) {
        Set<Map.Entry<String, Object>> entries = data.entrySet();
        String[] split = sinkColumns.split(",");
        List<String> splits = Arrays.asList(split);

        entries.removeIf(column -> !splits.contains(column.getKey()));
    }

    @Override
    public void processBroadcastElement(String value, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {

        // 获取配置表信息并解析
        JSONObject jsonObject = JSON.parseObject(value);
        TableProcess tableProcess = JSON.parseObject(jsonObject.getString("after"), TableProcess.class);

        // 校验表是否存在，如果不存在，则创建表
        checkTable(tableProcess.getSinkTable(),
                tableProcess.getSinkPk(),
                tableProcess.getSinkColumns(),
                tableProcess.getSinkExtend());

        // 将数据写入状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        broadcastState.put(tableProcess.getSourceTable(), tableProcess);

    }

    //校验表是否存在,如果不存在,则创建表
    //create table is not exists db.tn(id varchar primary key,name varchar,sex varchar) sinkExtend
    private void checkTable(String sinkTable, String sinkPk, String sinkColumns, String sinkExtend) {
        PreparedStatement preparedStatement = null;

        try {
            if (sinkPk == null || "".equals(sinkPk)){
                sinkPk = "id";
            }
            if (sinkExtend == null){
                sinkExtend = "";
            }

            StringBuilder sql = new StringBuilder("create table if not exists ")
                    .append(GmallConfig.HBASE_SCHEMA)
                    .append(".")
                    .append(sinkTable)
                    .append("(");

            String[] columns = sinkColumns.split(",");

            for (int i = 0; i < columns.length; i++) {
                String column = columns[i];

                // 判断当前字段是否为主键
                if (sinkPk.equals(column)) {
                    sql.append(column)
                            .append(" varchar primary key");
                }else {
                    sql.append(column)
                            .append(" varchar");
                }

                if (i < columns.length - 1){
                    sql.append(",");
                }
            }

            //拼接拓展字段
            sql.append(")")
               .append(sinkExtend);

            // 打印建表操作
            System.out.println(sql);

            // 建表
            preparedStatement = connection.prepareStatement(sql.toString());
            preparedStatement.execute();
        } catch (SQLException e) {
            e.printStackTrace();
            throw new RuntimeException("创建表" + sinkTable + "失败！！！");
        } finally {
            if (preparedStatement != null) {
                try {
                    preparedStatement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
